package org.code.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.code.common.Constant;
import org.code.util.ByteStringConverter;
import org.code.util.ThreadUtil;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

/**
 * @author xiao
 * <p>
 * 异步发送消息
 */
public class Async {

    public static final String TOPIC = "async-topic";

    /**
     * 消息生产者
     */
    @Test
    public void asyncProducer() throws Exception {
        // 创建消息生产者对象
        DefaultMQProducer producer = new DefaultMQProducer("async-producer-group");
        // 设置NameServer地址
        producer.setNamesrvAddr(Constant.NAME_SERVER);
        // 启动生产者
        producer.start();
        // 创建消息对象
        Message message = new Message(TOPIC, ByteStringConverter.stringToByteArray("async message"));
        List<Message> messageList = Arrays.asList(new Message(TOPIC, ByteStringConverter.stringToByteArray("async message 1")),
                new Message(TOPIC, ByteStringConverter.stringToByteArray("async message 2")),
                new Message(TOPIC, ByteStringConverter.stringToByteArray("async message 3")));
        // 发送消息
        producer.send(messageList, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("消息发送成功, sendStatus: %s\n", sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("消息发送失败, exception: %s\n", e.getMessage());
            }
        });
        // 挂起jvm 因为回调是异步的不然测试不出来
        ThreadUtil.read();
        // 关闭生产者
        producer.shutdown();
    }

    /**
     * 消息消费者
     */
    @Test
    public void asyncConsumer() throws Exception {
        // 创建消息消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("async-consumer-group");
        // 设置NameServer地址
        consumer.setNamesrvAddr(Constant.NAME_SERVER);
        // 设置消费者订阅的Topic
        consumer.subscribe(TOPIC, "*");
        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
                System.out.printf("线程名称: %s, 消费消息: %s\n",
                        ThreadUtil.getCurrentThreadName(),
                        ByteStringConverter.byteArrayToString(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        // 阻塞
        ThreadUtil.read();
    }

}
